/*
 * RadioStreamer.cpp
 *
 *  Created on: Dec 15, 2011
 *      Author: shaohong
 */

#include <arpa/inet.h>
#include <errno.h>
#include <limits.h>
#include <mpg123.h>
#include <netdb.h>
#include <netinet/in.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
#include <sndfile.h>

#include "RadioStreamer.h"
#include "HttpHelper.h"
#include "ParsingUtils.h"
#include "TranscoderUtils.h"
#include "RadioStation.h"

#include <sstream>

using namespace std;

static AppLogger * logger = AppLogger::getInstance("RadioStreamer");

namespace nmps {

static char _httpReponseBuffer[64 * 1024];

static void cleanup(mpg123_handle *mh) {
	/* It's really to late for error checks here;-) */
	mpg123_close(mh);
	mpg123_delete(mh);
	mpg123_exit();
}

static int connectToStreamingServer(string & streamServerHost, int & streamserverPort) {
	logger->debug("http downloading from host: %s , port: %d",
			streamServerHost.c_str(), streamserverPort);
	// now, let's connect to the remote server and send our request
	int clientSockFD = HttpHelper::create_tcp_socket();
	struct sockaddr_in svrSocketAddr; // the socket address of the server
	/* clear and initialize socketaddr */
	bzero((char*) (&svrSocketAddr), sizeof(struct sockaddr_in));
	svrSocketAddr.sin_family = AF_INET;
	svrSocketAddr.sin_port = htons(streamserverPort);
	struct hostent *srvHostEntry = gethostbyname(streamServerHost.c_str());
	if (NULL == srvHostEntry) {
		logger->error("Can't get server's address\n");
		perror("Can't get server's address\n");
		exit(1);
	}
	// now we have server's IP address in the 32 bits format
	bcopy(srvHostEntry->h_addr_list[0], &(svrSocketAddr.sin_addr),
			sizeof(struct in_addr));
	char *srv_ascii_ip = inet_ntoa(svrSocketAddr.sin_addr);
	logger->debug("host: \"%s\" is resolved to %s", streamServerHost.c_str(),
			srv_ascii_ip);
	/*connecting to the server*/
	int connResult = connect(clientSockFD,
			(struct sockaddr*) (&svrSocketAddr), sizeof(svrSocketAddr));
	if (-1 == connResult) {
		string msg = "failed to connect server";
		logger->error(msg.c_str());
		perror(msg.c_str());
		exit(1);
	}
	logger->debug("Connected: server's official host name is %s",
			srvHostEntry->h_name);

	return clientSockFD;
}

/**
 * read n bytes from fd and write the data to dataBuffer
 */
static int readMoreData(int fd, unsigned char * buffer, int n2bRead) {

	unsigned char * startPos = buffer;

	int actuallyRead = read(fd, startPos, n2bRead);

	while (-1 != actuallyRead) {
		if (actuallyRead == n2bRead) {
			return EXIT_SUCCESS;
		}

		startPos = startPos + actuallyRead;

		n2bRead = n2bRead - actuallyRead;

		actuallyRead = read(fd, startPos, n2bRead);
	}

	return EXIT_FAILURE;
}

//skip the mp3 stream meta data
static void skipMp3StreamMetaData(int socketFD) {
	//next byte is the length of meta data
	char byteRead;
	read(socketFD, &byteRead, 1);
	int metaDataLength = 16 * (unsigned int) byteRead;

	if (metaDataLength > 0) {
//		fprintf(stderr, "metaDataLength = %d\n", metaDataLength);
		for (int k = 0; k < metaDataLength; k++) {
			read(socketFD, &byteRead, 1);
		}
	}
}

extern "C"
void * startMp3Downloading (void* arg) {
	logger->debug("Mp3Downloading Thread started!");
	Mp3DownloadingThreadParams * mp3DownloadingThreadParams =
			(Mp3DownloadingThreadParams *)(arg);

	int metaint = 0;

	HttpRequest httpRequest;
	httpRequest.remoteHostName = HttpHelper::extractHost(mp3DownloadingThreadParams->streamServerLocation);
	httpRequest.URI = HttpHelper::extractPath(mp3DownloadingThreadParams->streamServerLocation);
	httpRequest.remoteHostPort = HttpHelper::extractPort(mp3DownloadingThreadParams->streamServerLocation);;

	//setup TCP connection with the remote streaming server sth like: http://69.166.45.47:8000
	int clientSockFD = connectToStreamingServer(httpRequest.remoteHostName, httpRequest.remoteHostPort);

	string httpRequestStr = HttpHelper::createHttpRequestStr(httpRequest, true);

	// now let's send HTTP request to streaming server
	send(clientSockFD, httpRequestStr.c_str(), httpRequestStr.length(), 0);
	logger->debug("sending out HTTP request:\n%s", httpRequestStr.c_str());

	unsigned int totalSize = 0;
	memset(_httpReponseBuffer, 0, sizeof(_httpReponseBuffer));

	// read from the socket, until we see two consecutive 0d0a0d0a
	/* Save the response to httpReponseBuffer */
	char buffer[1024];
	memset(buffer, 0, sizeof(buffer));
	int bufsize = read(clientSockFD, buffer, sizeof(buffer));
	bool streamingContentStarted = false;

	unsigned char * mp3FrameReadBuffer = NULL;

	//read the response and try to find the separator between header and body - "\r\n\r\n" in the buffer
	while ((0 != bufsize) && (-1 != bufsize)) {

		// trying to find the separator between header and body - "\r\n\r\n" in the buffer
		bool foundEmptyLine = false;
		int contentStartPos = -1;
		const char * crlfcrlf = "\r\n\r\n";
		char * pch = strstr(buffer, crlfcrlf);
		if (NULL != pch) {
			foundEmptyLine = true;
			contentStartPos = (pch - buffer) + 4;
//			fprintf(stderr, "found 0a0d0a0d!\n");
		}

		if (!foundEmptyLine) {
			// copy the whole read buffer
			memcpy(_httpReponseBuffer + totalSize, buffer, bufsize);
			totalSize += bufsize;

			//and continue to read
			memset(buffer, 0, sizeof(buffer));
			bufsize = read(clientSockFD, buffer, sizeof(buffer));
			continue;

		} else {
			// copy only the bytes before contentStartPos
			memcpy(_httpReponseBuffer + totalSize, buffer, contentStartPos);
			totalSize += contentStartPos;
			streamingContentStarted = true;

			logger->debug("Response header has %d bytes", totalSize);

			logger->debug("HTTP response received:================\n");
			logger->debug(_httpReponseBuffer);

			//extract the icy-metaint value
			metaint = HttpHelper::extractIcyMetaInt(_httpReponseBuffer);
			if (metaint <= 0) {
				logger->error("couldn't parse icy-metaint :(");
				exit(1);
			}

			logger->debug("icy-metaint = %d", metaint);

			//we've got meta-int, now lets receive the mp3 content!

			//allocate mp3FrameReadBuffer to be the size of the metaint
			mp3FrameReadBuffer = (unsigned char *) malloc(metaint);

			if (NULL == mp3FrameReadBuffer) {
				logger->error("couldn't parse icy-metaint :(");
				exit(1);
			}

			//copy the starting part of the content to mp3FrameReadBuffer
			int contentReadSoFar = (bufsize - contentStartPos);
			memcpy(mp3FrameReadBuffer, buffer + contentStartPos,
					contentReadSoFar);

			//and read further to fill the mp3FrameReadBuffer;
			int contentToBeRead = metaint - contentReadSoFar;
			if (EXIT_FAILURE
					== readMoreData(clientSockFD,
							mp3FrameReadBuffer + contentReadSoFar,
							contentToBeRead)) {
				logger->error("failed to read mp3 content!:(");
				exit(1);
			}

			// this is the end of looking at HTTP response stuff.
			break;

		}

	} // end of handling the initial part of the HTTP response stream

	// starting from here, let's take care of mp3 stream
	long framesRead = 1;

	RadioStreamer * radioStreamer = mp3DownloadingThreadParams->radioStreamer;
	while(!radioStreamer->isStoped()){
		//skip the mp3 stream meta data
		skipMp3StreamMetaData(clientSockFD);

		//continue to read to mp3ReadBuffer;
		if (EXIT_FAILURE
				== readMoreData(clientSockFD,
						mp3FrameReadBuffer, metaint)){
			logger->error("failed to read mp3 content!:(");
			exit(1);
		}
		framesRead ++;

		//save the MP3 frames read into streamer stats
		radioStreamer->stats.increaseMp3FramesReceived();


//		//skip the first 5 frames
//		if (framesRead < 5) {
//			continue;
//		}
		//write it to the pipe so that the mp3Transcoding Thread can transcode it
		write(mp3DownloadingThreadParams->writeFD, mp3FrameReadBuffer, metaint);
//		fprintf(stderr, "m");
//		fprintf(stderr, "mp3framesRead = %ld", framesRead);
	}


	close(clientSockFD);
	free(mp3FrameReadBuffer);

	logger->debug("Mp3Downloading Thread finished!");
	return 0;
}

void * startTranscoding (void* arg)
{
	logger->debug("Transcoding Thread started!");

	Mp3TranscodingThreadParams * threadParams =
			static_cast<Mp3TranscodingThreadParams*>(arg);

	int instreamFD = threadParams->readFD;
	RadioStreamer * radioStreamer = threadParams->radioStreamer;

	SNDFILE* sndfile = NULL;
	SF_INFO sfinfo;

	mpg123_handle *mh = NULL;
	unsigned char* buffer = NULL;
	size_t buffer_size = 0;
	size_t done = 0;
	int channels = 0, encoding = 0;
	long rate = 0;
	int err = MPG123_OK;

	err = mpg123_init();

	if (err != MPG123_OK || (mh = mpg123_new(NULL, &err)) == NULL
	/* Let mpg123 work with the file, that excludes MPG123_NEED_MORE messages. */
	|| mpg123_open_fd(mh, instreamFD) != MPG123_OK) {
		fprintf(stderr, "Trouble with mpg123: %s\n",
				mh == NULL ? mpg123_plain_strerror(err) : mpg123_strerror(mh));
		cleanup(mh);
		exit(EXIT_FAILURE);
	}

	// force the rate to be 8000
	err = mpg123_param(mh, MPG123_FORCE_RATE, 8000, 0);

	// disable some of the mpg123 console output
	err = mpg123_param(mh, MPG123_VERBOSE, 0, 0);

	mpg123_format_none(mh);
	mpg123_format(mh, 8000, MPG123_MONO, MPG123_ENC_SIGNED_16);

	mpg123_getformat(mh, &rate, &channels, &encoding);

	fprintf(stderr, "rate=%ld, channels=%d, encoding=%d\n", rate, channels,
			encoding);
	logger->info("transcoding: rate=%ld, channels=%d, encoding=%d\n", rate,
			channels, encoding);

	bzero(&sfinfo, sizeof(sfinfo));
	sfinfo.samplerate = rate;
	sfinfo.channels = channels;
	sfinfo.format = SF_FORMAT_RAW | SF_FORMAT_PCM_16;
	char waveFileName[200];
	sprintf(waveFileName, "radio_%d.raw", radioStreamer->getRadioChannel().channelNumber);
	sndfile = sf_open(waveFileName, SFM_WRITE, &sfinfo);
	if (sndfile == NULL) {
		fprintf(stderr, "Cannot open output file!\n");
		cleanup(mh);
		exit(EXIT_FAILURE);
	}

	/* Buffer could be almost any size here, mpg123_outblock() is just some recommendation.
	 Important, especially for sndfile writing, is that the size is a multiple of sample size. */
	buffer_size = mpg123_outblock(mh);
	buffer = (unsigned char *) malloc(buffer_size);

	do {
		err = mpg123_read(mh, buffer, buffer_size, &done);

		sf_write_raw(sndfile, buffer, done);

		if (done == 0) {
			continue;
		}

		radioStreamer->transcodingStarted = true;
		//save the time we send out the first real RTP packets
		if (! radioStreamer->stats.transcodedReady) {
			radioStreamer->stats.markTranscodedAudioReady();
		}

		unsigned char tempBuf[3000];

		//convert PCM16 to pcma and save pcma in tempBuf
		for (int i = 0; i < (int) done / 2; i++) {
			int pcm16 = *((int16_t *) buffer + i);

			tempBuf[i] = s16_to_alaw(pcm16);
		}

		// save the PCMA data to rtpStreamer's buffer
		pthread_mutex_t rtpstreamerMutex = radioStreamer->getRtpStreamerMutex();
		pthread_mutex_lock(&rtpstreamerMutex);
		PcmPlayoutBuffer * audioPlayoutBuffer =
				radioStreamer->getRtpStreamer()->getPlayoutBuffer();
		audioPlayoutBuffer->enQueuePcmDataBlock((int) (done / 2), tempBuf);
		pthread_mutex_unlock(&rtpstreamerMutex);

	} while ((err == MPG123_OK) && !radioStreamer->isStoped());

	if (err != MPG123_DONE) {
		fprintf(
				stderr,
				"Warning: Decoding ended prematurely because: %s\n",
				err == MPG123_ERR ?
						mpg123_strerror(mh) : mpg123_plain_strerror(err));
	}

	cleanup(mh);

	if (err == MPG123_ERR) {
		exit(EXIT_FAILURE);
	}

	logger->debug("Transcoding Thread finished!");
	return 0;
}

/**
 * RTP Sending thread.
 * It starts a timer and every 20ms, it will get 160 bytes from the “AudioPlayoutBuffer”
 * and send the audio data to remote client via RTP.
 */
void * startRTPStreaming(void *arg) {

	logger->debug("RTPStreaming Thread started!");

	const long MILLION = 100000L;
	RptStreamingThreadParams * threadParams =
			static_cast<RptStreamingThreadParams*>(arg);

	struct timeval tBefore, tAfter;
	long timedif;

	struct timeval tv;
	tv.tv_sec = 0;
	tv.tv_usec = 20 * 1000; //20 ms

	RadioStreamer * radioStreamer = threadParams->radioStreamer;
	RtpStreamer * rtpStreamer = radioStreamer->getRtpStreamer();
	while (!threadParams->radioStreamer->isStoped()) {

		select(0, NULL, NULL, NULL, &tv);

		gettimeofday(&tBefore, NULL);

		if (rtpStreamer->getPlayoutBuffer()->getNumberOfBlocks() == 0){
			continue;
		}
		pthread_mutex_t rtpstreamerMutex = threadParams->radioStreamer->getRtpStreamerMutex();
		pthread_mutex_lock(&rtpstreamerMutex);

		radioStreamer->stats.increasePcmFramesCreated();
		rtpStreamer->sendDataOverRTP(radioStreamer->isPaused());

		pthread_mutex_unlock(&rtpstreamerMutex);

		gettimeofday(&tAfter, NULL);

		//adjust for the time spent sending the packet
		timedif = (tAfter.tv_sec - tBefore.tv_sec) * MILLION + (tAfter.tv_usec - tBefore.tv_usec);
		tv.tv_sec = 0;
		tv.tv_usec = 20 * 1000 - timedif;
		if (tv.tv_usec <=0){
			fprintf(stderr, "-");
			tv.tv_usec = 0;
		}

	}

	return 0;
}


RadioStreamer::RadioStreamer(RadioChannel radioChannel):stats(&radioChannel) {

	logger->debug("RadioStreamer created for channel: %d", radioChannel.channelNumber);

	this->radioChannel = radioChannel;

	rtpStreamerMutex = PTHREAD_MUTEX_INITIALIZER;

	rtpStreamer = new RtpStreamer();
	rtpStreamer->initialize();

	//initialize the statistics for this radio streamer
	stats.initialize();

	paused = false;

	//initially streaming is not started.
	streamingStarted = false;

	stopped = false;

	transcodingStarted = false;

}

RadioStreamer::~RadioStreamer() {

}

bool RadioStreamer::isPaused() {
	return paused;
}

bool RadioStreamer::isStoped() {
	return stopped;
}

bool RadioStreamer::isStreamingStarted() {
	return streamingStarted;
}

RadioChannel RadioStreamer::getRadioChannel() {
	return radioChannel;
}

pthread_t RadioStreamer::getRadioStreamerThread() {
	return downloadingThread;
}

string RadioStreamer::toString() {
	stringstream oss;
	oss << "{RadioStreamer: " << radioChannel.toString() << "}";
	return oss.str();
}

//checks whether the actual transcoding is started
bool RadioStreamer::isTranscodingStarted() {
	return transcodingStarted && isStreamingStarted();
}

// start the mp3 downloading, transcoding, streaming, pipeline
void RadioStreamer::startStreaming() {

	this->stats.setRadioChannel(&radioChannel);
	this->stats.initialize();

	logger->debug("%s startStreaming()", toString().c_str());

	if (isStreamingStarted()) {
		logger->debug("%s streaming already started!", toString().c_str());

		return;
	}

	RadioStation::instance()->addActiveRadioChannel(this);

	this->stopped = false;
	this->streamingStarted = true;
	this->paused = false;

	//save the time before we send out HTTP request
	stats.markFirstHttpRequestSent();

	// retrieve the playlist according to the playlistURL
	//extract the host part from the playlistURI
	HttpRequest httpRequest;

	httpRequest.remoteHostName = HttpHelper::extractHost(radioChannel.playlistURL);// "yp.shoutcast.com";
	httpRequest.URI = HttpHelper::extractPath(radioChannel.playlistURL); //"/sbin/tunein-station.pls?id=1926281";
	httpRequest.remoteHostPort = 80;

	HttpResponse response;
	HttpHelper::retrieveURIContent(httpRequest, response);

	logger->debug("HTTP response received:================\n");
	logger->debug(response.data);

	//get the HTTP mediastream file location e.g. "http://69.166.45.47:8000"
	string playfileLocation = ParsingUtils::getPlayFileLocationFromHttpResponse(response.data);
	logger->debug("%s : play file location is %s", toString().c_str(), playfileLocation.c_str());

//	HttpStreamerClient httpStreamerClient = HttpStreamerClient(playfileLocation);
//	httpStreamerClient.startStreamingDownload();

	int fd[2];
	if (pipe(fd) == -1){
		perror("Failed to create the pipe");
		exit(-1);
	}

	// fire off the downloading thread
	Mp3DownloadingThreadParams * mp3DownloadingThreadParams = new Mp3DownloadingThreadParams();
	mp3DownloadingThreadParams->writeFD = fd[1];
	mp3DownloadingThreadParams->streamServerLocation = playfileLocation;
	mp3DownloadingThreadParams->radioStreamer = this;

	pthread_create(&downloadingThread, NULL, startMp3Downloading, mp3DownloadingThreadParams);

	// fire off the transcoding thread
	Mp3TranscodingThreadParams * transCodingThreadParams = new Mp3TranscodingThreadParams;
	transCodingThreadParams->readFD = fd[0];
	transCodingThreadParams->radioStreamer = this;

	pthread_create(&transcodingThread, NULL, startTranscoding, transCodingThreadParams);

	// fire off the RTP streaming thread
	RptStreamingThreadParams * rptStreamingThreadParams = new RptStreamingThreadParams();
	rptStreamingThreadParams->radioStreamer = this;
	pthread_create(&rtpStreamingThread, NULL, startRTPStreaming, rptStreamingThreadParams);
}

// temporarily pause the RTP streamer streaming real audio to recipients
void RadioStreamer::pauseStreaming() {
	logger->debug("%s pauseStreaming()", toString().c_str());
	this->paused = true;

}

// let the RTP streamer continue streaming audio to recipients
void RadioStreamer::resumeStreaming() {
	logger->debug("%s resumeStreaming()", toString().c_str());
	this->paused = false;
}

// stop the whole streaming related threads
void RadioStreamer::stopStreaming() {
	logger->debug("%s stopStreaming()", toString().c_str());

	if (isStoped()) {
		return;
	}

	stopped = true;
	streamingStarted = false;
	transcodingStarted = false;

	pthread_join(downloadingThread, NULL);
	pthread_join(transcodingThread, NULL);
	pthread_join(rtpStreamingThread, NULL);

	//clear the audio playout buffer of the RTP streamer
	this->rtpStreamer->clearAudioPlayoutBuffer();

	// save previous statics to history
	this->stats.markStationStopTime();
	writeStatsToHistory();

	RadioStation::instance()->removeActiveRadioChannel(this);

}

/**
 * switch the current radio streamer to listen to new Channel
 */
void RadioStreamer::switchChannel(RadioChannel newChannel) {
	this->stopStreaming();

	this->radioChannel = newChannel;

	this->startStreaming();
}

// add the recipient identified by (IP, Port) couple
void RadioStreamer::addRecipient(string clientIP, int clientPort) {
	logger->debug("%s addRecipient(%s, %d)", toString().c_str(), clientIP.c_str(), clientPort);
	pthread_mutex_lock(&rtpStreamerMutex);
	this->rtpStreamer->addRecipient(clientIP, clientPort);
	pthread_mutex_unlock(&rtpStreamerMutex);
}

void RadioStreamer::addRecipients(list<pair<string, int>  > & recipientList) {
	list<pair<string, int>  >::iterator itr = recipientList.begin();

	while(itr != recipientList.end()) {
		addRecipient(itr->first, itr->second);
		itr ++;
	}
}
const list< pair<string, int> > * RadioStreamer::getRecipientList() {
	return rtpStreamer->getRecipientList();
}


// remove the streaming recipient identified by (IP, Port) couple
void RadioStreamer::removeReceipient(string clientIP, int clientPort) {
	logger->debug("%s removeReceipient(%s:%d)", toString().c_str(), clientIP.c_str(), clientPort);

	pthread_mutex_lock(&rtpStreamerMutex);
	this->rtpStreamer->removeRecipient(clientIP, clientPort);
	pthread_mutex_unlock(&rtpStreamerMutex);
}

RtpStreamer *RadioStreamer::getRtpStreamer() const {
	return rtpStreamer;
}

pthread_mutex_t RadioStreamer::getRtpStreamerMutex() {
	return rtpStreamerMutex;
}

void RadioStreamer::setRtpStreamer(RtpStreamer *rtpStreamer) {
	this->rtpStreamer = rtpStreamer;
}

void RadioStreamer::setRtpStreamerMutex(pthread_mutex_t rtpStreamerMutex) {
	this->rtpStreamerMutex = rtpStreamerMutex;
}

void RadioStreamer::writeStatsToHistory() {
	RadioStation::instance()->saveRadioChannelHistory(stats.toString());
}



} /* namespace nmps */
